package com.ylkj.bus.app.report;

import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RdsUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.text.MessageFormat;
import java.util.Arrays;
import java.util.List;

/**
 * @Author: wengzx
 * @Date: 2025/3/23
 * @Desc: 处理t_iotp_container_app表中数据(先删除后插入)
 */
@Slf4j
public class UnpackContainerAppDeleteAndInsert extends ProcessFunction<List<ContainerAppDO>,List<ContainerAppDO>> {
    //基础删除SQL模板
    private static final String BASE_DELETE_SQL = "DELETE FROM {0} WHERE container_id = ? ";

    private static final String INSERT_SQL =
            "INSERT INTO {0} (`id`, `container_id`, `app_id`, `option_status`, `serivce_status`, `enable_status`, `belong_id`, `source`, `app_hash`, create_time, update_time) " +
                    "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, now(), now()) " +
                    "ON DUPLICATE KEY UPDATE " +
                    "`app_id` = VALUES(`app_id`), " +
                    "`option_status` = VALUES(`option_status`), " +
                    "`serivce_status` = VALUES(`serivce_status`), " +
                    "`enable_status` = VALUES(`enable_status`), " +
                    "`belong_id` = VALUES(`belong_id`), " +
                    "`source` = VALUES(`source`), " +
                    "`app_hash` = VALUES(`app_hash`), " +
                    "`create_time` = VALUES(`create_time`), " +
                    "`update_time` = now()";

    // 数据库配置和表名
    private final String tableName;
    private final RdsConfig rdsConfig;
    private RdsUtils rdsUtils;

    // 格式化后的基础SQL语句
    private String baseSql;

    private String insertOnDuplicateSql;

    private static final int DEADLOCK_ERROR_CODE = 1213;

    //构造函数，用于初始化
    public UnpackContainerAppDeleteAndInsert(String tableName , RdsConfig rdsConfig) {
        this.rdsConfig = rdsConfig;
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        rdsUtils = new RdsUtils(rdsConfig);
        baseSql = MessageFormat.format(BASE_DELETE_SQL, tableName);
        insertOnDuplicateSql = MessageFormat.format(INSERT_SQL, tableName);
    }

    @Override
    public void processElement(List<ContainerAppDO> containerAppDOs, ProcessFunction<List<ContainerAppDO>, List<ContainerAppDO>>.Context ctx, Collector<List<ContainerAppDO>> out) throws Exception {
        if(containerAppDOs ==null || containerAppDOs.isEmpty()){
            return;
        }
        ContainerAppDO containerAppDO = containerAppDOs.get(0);
        String containerId = containerAppDO.getContainerId();
        int retryCount = 0;
        while (true){
            try (Connection connection = rdsUtils.getConnection()) {
                connection.setAutoCommit(false); //开启事务
                try {
                    //删除旧的记录
                    boolean existsAndDeleted = deleteDeviceIdAndContainer(connection, containerId);
                    if (existsAndDeleted) {
                        log.info("已删除旧记录,containerId: {}",  containerId);
                    }
                    // 插入新记录（无论是否存在）
                    int insertedRows = insertBatchData(connection, containerAppDOs);
                    log.info("成功插入{}条新记录，containerId: {}", insertedRows, containerId);
                    connection.commit(); // 提交事务
                    break; // 成功退出循环
                } catch (SQLException e) {
                    if (e.getErrorCode() == DEADLOCK_ERROR_CODE && retryCount < 3) {
                        retryCount++;
                        log.warn("死锁重试 {} 次: {}", retryCount, e.getMessage());
                        continue; // 重试事务
                    } else {
                        connection.rollback();
                        log.error("数据库操作失败，已回滚事务。错误信息: {}", e.getMessage(), e);
                        throw e;
                    }
                }
            }
        }
    }

    /**
     * 删除指定 container_id 的数据
     * @param connection
     * @param containerId
     * @return
     * @throws SQLException
     */
    private boolean deleteDeviceIdAndContainer(Connection connection, String containerId) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(baseSql)) {
            stmt.setString(1, containerId);
            int affectedRows = stmt.executeUpdate();
            return affectedRows > 0; // 如果删除了数据，返回 true
        }
    }

    /**
     *  批量插入或更新数据
     * @param connection
     * @param containerAppDOs
     * @return
     * @throws SQLException
     */
    private int insertBatchData(Connection connection, List<ContainerAppDO> containerAppDOs) throws SQLException {
        try (PreparedStatement stmt = connection.prepareStatement(insertOnDuplicateSql)) {
            int batchSize = 0;
            for (ContainerAppDO item : containerAppDOs) {
                try {
                    stmt.setString(1, item.getId());
                    stmt.setString(2, item.getContainerId());
                    stmt.setString(3, item.getAppId());
                    stmt.setString(4, item.getOptionStatus());
                    stmt.setString(5, item.getSerivceStatus());
                    stmt.setString(6, item.getEnableStatus());
                    stmt.setString(7, item.getBelongId());
                    stmt.setLong(8, item.getSource());
                    stmt.setString(9, item.getAppHash());
                    stmt.addBatch();
                    if (++batchSize % 5 == 0) { // 每5条执行一次批量插入
                        stmt.executeBatch();
                    }
                }  catch (SQLException e) {
                    log.error("插入或更新记录时出错，跳过。ID：{}，错误：{}", item.getId(), e.getMessage(), e);
                }
            }
            stmt.executeBatch(); // 插入剩余数据
            return batchSize;
        }
    }


    @Override
    public void close() throws Exception {
        rdsUtils.close();
        super.close();
    }


}
